package io.terminus.paas.profiler.monitor

import io.terminus.paas.profiler.model.TracingRecord
import org.influxdb.InfluxDB
import org.influxdb.InfluxDBFactory
import org.influxdb.dto.BatchPoints
import org.influxdb.dto.Point
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

/**
 * Created by gengrong on 16/5/18.
 */
class InFluxDBMonitor : AbstractMonitor() {

    val threadPool = Executors.newCachedThreadPool();

    val dataBase = "pampas_profiler_monitor";

    val influxDB = InfluxDBFactory.connect("http://192.168.0.251:8086", "root", "root");

    private var queue: MutableList<TracingRecord> = mutableListOf()

    private val thresholdTime = 10000

    private var lastSyncTime = System.currentTimeMillis();

    override fun recording(record: TracingRecord) {
        queue.add(record)
        val time = System.currentTimeMillis() - lastSyncTime;
        if (time > thresholdTime) {
            threadPool.execute({
                sync()
            })
        }
    }

    private fun sync() {
        lastSyncTime = System.currentTimeMillis();
        val list = queue;
        queue = mutableListOf()
        val batchPoints = buildBatch()
        list.forEach { record ->
            val point = buildPoint(record);
            batchPoints.point(point)
        }
        influxDB.write(batchPoints)
    }

    private fun buildPoint(record: TracingRecord): Point {
        val executionTime = record.endTime!! - record.startTime!!;
        val point = Point.measurement(record.interfaceName)
                .time(record.startTime!!, TimeUnit.MILLISECONDS)
                .tag("interfaceName", record.interfaceName)
                .tag("rootId", record.rootId)
                .tag("tracingId", record.tracingId)
                .addField("parentId", record.parentId ?: "")
                .addField("methodName", record.methodName)
                .addField("executionTime", executionTime)
                .addField("status", record.status.toString())
                .build();
        return point;
    }

    private fun buildBatch(): BatchPoints {
        influxDB.createDatabase(dataBase)
        val batchPoints = BatchPoints.database(dataBase)
                .retentionPolicy("default")
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        return batchPoints;
    }
}